package com.suxin.tomcat;

import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author Tang
 * @classname NioEndPoint
 * @description [  ]
 * @date 2021/7/3 15:16
 */
@Slf4j
public class NioEndPoint extends AbstractEndPoint {

    private int threadPriority = Thread.NORM_PRIORITY;
    private ServerSocketChannel serverSocket;


    protected int acceptorThreadCount = 1;

    public int getAcceptorThreadCount() {
        return acceptorThreadCount;
    }

    private Acceptor[] acceptors = null;

    protected class Acceptor extends AbstractEndPoint.Acceptor {

        @Override
        public void run() {
            while (true) {
                SocketChannel socket = null;
                try {
                    // todo 并发量过量问题
//                getAcquire();
                    log.info("{},线程等待接收连接中...", Thread.currentThread().getName());
                    socket = serverSocket.accept();
                    log.info("{},服务端收到新的连接: {}", Thread.currentThread().getName(), socket.hashCode());
                } catch (Exception e) {
//                releaseConnectionLatch();
                }
                if (!setSocketOptions(socket)) {
                    closeSocket(socket);
                }
            }
        }
    }

    private boolean setSocketOptions(SocketChannel socket) {
        try {
            // 非阻塞模式开启
            socket.configureBlocking(false);
            Socket sock = socket.socket();
            //setSocketProperties(sock);
            getPoller0().register(socket);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return true;
    }

    void setSocketProperties(Socket socket) throws Exception {
        socket.setReceiveBufferSize(0);
        socket.setSendBufferSize(0);
        socket.setKeepAlive(true);
    }

    private void closeSocket(SocketChannel socket) {
        try {
            socket.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    public static final int OP_REGISTER = 0x100;

    protected static class PollerEvent extends AbstractEndPoint.PollerEvent {

        NioSocketWrapper nioSocketWrapper;
        // 当前感兴趣的事件
        private int interestOps;

        public PollerEvent(NioSocketWrapper nioSocketWrapper, int ops) {
            this.nioSocketWrapper = nioSocketWrapper;
            this.interestOps = ops;
        }

        @Override
        public void run() {
            SocketChannel socketChannel = nioSocketWrapper.getSocket();
            Selector selector = nioSocketWrapper.getPoller().getSelector();
            try {
                // todo 开始注册时会执行事件
                if (interestOps == OP_REGISTER) {
                    socketChannel.register(selector,
                            SelectionKey.OP_READ, nioSocketWrapper);
                    log.info("{},为当前连接:{}注册读事件...", Thread.currentThread().getName(), socketChannel.hashCode());
                } else {
                    // todo 动态改变感兴趣的事件
                    SelectionKey selectionKey = socketChannel.keyFor(selector);
                    selectionKey.interestOps(selectionKey.interestOps() | interestOps);
                }
            } catch (ClosedChannelException e) {
                e.printStackTrace();
            }
        }
    }

    public static class SocketProcessor extends SocketProcessorBase {

        private NioSocketWrapper nioSocketWrapper;
        SocketEvent socketEvent;

        public SocketProcessor(NioSocketWrapper nioSocketWrapper, SocketEvent socketEvent) {
            this.nioSocketWrapper = nioSocketWrapper;
            this.socketEvent = socketEvent;
        }

        @Override
        protected void doRun() {
            SocketChannel socket = this.nioSocketWrapper.getSocket();
            int handshake = -1;

            String readDate = nioSocketWrapper.getSocketBufferHandler().readDate(socket);

            // todo 业务处理
            log.info("{}-业务线程,处理当前连接-{},业务数据-{}", Thread.currentThread().getName(),
                    socket.hashCode(), readDate);

            if (handshake == SelectionKey.OP_READ) {
                nioSocketWrapper.registerReadInterest();
            }

            if (handshake == SelectionKey.OP_WRITE) {
                nioSocketWrapper.registerWriterInterest();
            }

        }
    }


    private int pollerThreadCount = Math.min(2, Runtime.getRuntime().availableProcessors());

    public int getPollerThreadCount() {
        return pollerThreadCount;
    }

    private Poller[] pollers = null;
    private AtomicInteger pollerRotater = new AtomicInteger(0);

    private long selectorTimeout = 1000;

    public long getSelectorTimeout() {
        return selectorTimeout;
    }

    protected class Poller extends AbstractEndPoint.Poller {

        // 每一个 poller 都有一个 selector
        private Selector selector;
        // 等待事件处理的队列
        private final BlockingQueue<PollerEvent> events = new ArrayBlockingQueue<>(30);

        public Poller() {
            try {
                // 初始化selector
                selector = Selector.open();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        @Override
        public void run() {
            while (true) {
                // 注册事件
                events0();
                try {
                    int keyCount = selector.select(getSelectorTimeout());
                    Iterator<SelectionKey> iterator = keyCount > 0 ? selector.selectedKeys().iterator() : null;
                    while (iterator != null && iterator.hasNext()) {
                        SelectionKey selectionKey = iterator.next();
                        iterator.remove();
                        NioSocketWrapper nioSocketWrapper = (NioSocketWrapper) selectionKey.attachment();
                        if (nioSocketWrapper != null) {
                            log.info("{},处理当前连接...{}", Thread.currentThread().getName(), nioSocketWrapper.getSocket().hashCode());
                            // 处理连接
                            processKey(selectionKey, nioSocketWrapper);
                        }
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }

        private void processKey(SelectionKey selectionKey, NioSocketWrapper nioSocketWrapper) {
            if (selectionKey.isReadable() || selectionKey.isWritable()) {
                // 注销这次事件，防止多次触发
                unreg(selectionKey, nioSocketWrapper,selectionKey.readyOps());
                if (selectionKey.isReadable()) {
                    if (!processSocket(nioSocketWrapper, SocketEvent.OPEN_READ)) {
                        //
                    }
                }
                if (selectionKey.isWritable()) {
                    if (!processSocket(nioSocketWrapper, SocketEvent.OPEN_WRITE)) {
                        //
                    }
                }
            }
        }

        private void unreg(SelectionKey selectionKey, NioSocketWrapper nioSocketWrapper, int readyOps) {
            reg(selectionKey, nioSocketWrapper, selectionKey.interestOps() & (~readyOps));
        }

        private void reg(SelectionKey selectionKey, NioSocketWrapper nioSocketWrapper, int intOps) {
            selectionKey.interestOps(intOps);
        }

        /**
         * 处理socket
         *
         * @param nioSocketWrapper
         * @param socketEvent
         * @return
         */
        private boolean processSocket(NioSocketWrapper nioSocketWrapper, SocketEvent socketEvent) {
            // 业务线程
            Executor executor = getExecutor();
            SocketProcessor socketProcessor = new SocketProcessor(nioSocketWrapper, socketEvent);
            executor.execute(socketProcessor);
            return true;
        }

        /**
         * 消费队列注册
         */
        private void events0() {
            PollerEvent pollerEvent = null;
            for (int i = 0, size = events.size(); i < size && (pollerEvent = events.poll()) != null; ) {
                pollerEvent.run();
            }
        }

        public void register(SocketChannel socket) {
            NioSocketWrapper nioSocketWrapper = new NioSocketWrapper(socket, this,
                    NioEndPoint.this);
            PollerEvent pollerEvent = new PollerEvent(nioSocketWrapper, OP_REGISTER);
            // 添加到队列
            addEvent(pollerEvent);
        }

        private void addEvent(PollerEvent pollerEvent) {
            events.offer(pollerEvent);
        }

        public Selector getSelector() {
            return selector;
        }

        public BlockingQueue<PollerEvent> getEvents() {
            return events;
        }

        public void add(NioSocketWrapper nioSocketWrapper, int opst) {
            addEvent(new PollerEvent(nioSocketWrapper, opst));
        }
    }

    private Poller getPoller0() {
        int idx = Math.abs(pollerRotater.getAndIncrement() % pollers.length);
        return pollers[idx];
    }

    protected static class NioSocketWrapper extends SocketWrapperBase {

        private SocketChannel socket;
        private Poller poller;
        private NioEndPoint nioEndPoint;

        public NioSocketWrapper(SocketChannel socket, Poller poller, NioEndPoint nioEndPoint) {
            this.socket = socket;
            this.poller = poller;
            this.nioEndPoint = nioEndPoint;
            this.socketBufferHandler = new SocketBufferHandler(1024, 1024);
        }

        public SocketChannel getSocket() {
            return socket;
        }

        public Poller getPoller() {
            return poller;
        }

        public NioEndPoint getNioEndPoint() {
            return nioEndPoint;
        }

        @Override
        public void registerReadInterest() {
            getPoller().add(this, SelectionKey.OP_READ);
        }

        @Override
        public void registerWriterInterest() {
            getPoller().add(this, SelectionKey.OP_WRITE);
        }
    }

    @Override
    public void bind() {
        try {
            serverSocket = ServerSocketChannel.open();
            // 连接接收是阻塞的
            serverSocket.configureBlocking(true);
            serverSocket.bind(new InetSocketAddress("127.0.0.1", 9999), 1024);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void startAcceptorThreads() {
        acceptors = new Acceptor[getAcceptorThreadCount()];
        for (int i = 0; i < acceptors.length; i++) {
            acceptors[i] = new Acceptor();
            String threadName = "Acceptor-" + i;
            Thread acceptorThread = new Thread(acceptors[i], threadName);
            acceptorThread.setPriority(threadPriority);
            acceptorThread.setDaemon(true);
            acceptorThread.start();
        }
    }

    public void initInternal() {
        bind();
    }

    public void startInternal() {
        // work executor
        if (getExecutor() == null) {
            createExecutor();
        }
        // 限流器
        initializeConnectionLatch();

        // 启动poller线程
        pollers = new Poller[getPollerThreadCount()];
        for (int i = 0; i < pollers.length; i++) {
            pollers[i] = new Poller();
            Thread pollerThread = new Thread(pollers[i], "poller-" + i);
            pollerThread.setPriority(threadPriority);
            pollerThread.setDaemon(true);
            pollerThread.start();
        }

        // 启动 acceptor 线程
        startAcceptorThreads();
    }

    public static void main(String[] args) {
        NioEndPoint nioEndPoint = new NioEndPoint();
        nioEndPoint.initInternal();
        nioEndPoint.startInternal();

        try {
            System.in.read();
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

}